#include "StreamServer.h"
#include "Acceptor.h"
#include "EventLoop.h"
#include "Manager.h"
#include "PlatLogger.h"
#include "StreamSocketChannel.h"
#include "SubStreamServer.h"
#include "Util.h"
#include "XnccAddress.h"
#include "XnccServer.h"
#include "XnccType.h"

namespace xncc {
namespace foundation {

    StreamServer::StreamServer(EventLoop*                   loop,
                               std::shared_ptr<XnccAddress> listenAddr,
                               std::string                  nameArg,
                               const std::set<std::string>& blacklist,
                               Option                       option)
        : XnccServer(blacklist),
          loop_(loop),
          ipPort_(listenAddr->toStr()),
          name_(std::move(nameArg)),
          acceptor_(new Acceptor(loop, listenAddr, option == kReusePort)),
          nextConnId_(1),
          nextSubserverIndex_(0)
    {
        if (loop_ == nullptr) {
            std::ostringstream oss;
            oss << " StreamServer engine  " << name_ << " must have a valid eventloop";
            throw shared::logic_exception(oss.str());
        }
    }

    StreamServer::~StreamServer() = default;

    void StreamServer::init()
    {
        loop_->registerUserFunc(shared::InnerMsgId::ON_PUB_IPC_MSG,
                                [this](const char* msg, uint32_t len, types::thread_id_t tid,
                                       types::connection_id_t connId) { handleIoMsg(msg, len, tid, connId); });

        acceptor_->init(blacklist_);
        acceptor_->setNewConnectionCallback(
            [this](int sockfd, std::shared_ptr<XnccAddress> peerAddr) { newConnection(sockfd, peerAddr); });
    }

    void StreamServer::start()
    {
        createSubServer();
        if (unlikely(acceptor_->listenning())) {
            throw shared::logic_exception("listen not allowed");
        }
        TRACE_INFO << "engine id:" << loop_->id() << " server started:"_s << ipPort_;
        loop_->runInLoop([this]() { acceptor_->listen(); });
    }

    void StreamServer::handleIoMsg(const char* msg, uint32_t len, types::thread_id_t tid, types::connection_id_t connId)
    {
        auto ptr = clientChs_[connId];
        if (ptr == nullptr) {
            TRACE_ERROR << "msg sender set error connid:"_s << connId;
            return;
        }
        ptr->sendMsg(msg, len);
    }

    void StreamServer::newConnection(int sockfd, std::shared_ptr<XnccAddress> peerAddr)
    {
        if (subServers_.size() > 0) {
            if (nextSubserverIndex_ == subServers_.size()) {
                TRACE_DEBUG << "engine id:"_s << loop_->id() << " main server index:"_s << nextSubserverIndex_;
                handleNewConnection(sockfd, peerAddr);
                nextSubserverIndex_ = 0;
            }
            else {
                auto subserver = subServers_[nextSubserverIndex_];
                TRACE_DEBUG << "engine id:"_s << loop_->id() << " subserver index:"_s << nextSubserverIndex_
                            << " subsever:" << subserver->getLoop()->id();
                subserver->acceptNewConnection(sockfd, peerAddr);
                ++nextSubserverIndex_;
            }
        }
        else {
            handleNewConnection(sockfd, peerAddr);
        }
    }

    void StreamServer::handleNewConnection(int sockfd, std::shared_ptr<XnccAddress> peerAddr)
    {
        const auto connId = shared::CommonBusi::generateConnId();
        // 构建消息
        const std::string peerAddrStr = peerAddr->toStr();
        QUEUE_MGR.pubIPCDataToBusiThread(peerAddrStr.c_str(), static_cast<uint32_t>(peerAddrStr.size()),
                                         shared::InnerMsgId::ON_CONNECTED, loop_->id(), connId);
        TRACE_INFO << " engine id:"_s << loop_->id() << " new connection, connid:"_s << connId << " peeraddr:"_s
                   << peerAddrStr;
        // 一个连接建立之后，投递连接信息到 关注消息的线程， 然后要把这个fd，放进epool循环中
        /// 这里创建的channel
        /// 是TCP连接的channel，断开连接时需要释放内存的，使用智能指针，但是连接断开的时候，只要erase了这个指针，内存才能被释放
        auto chPtr         = std::make_shared<StreamSocketChannel>(loop_, sockfd, connId);
        clientChs_[connId] = chPtr;
        chPtr->init();
        chPtr->setCloseCallback([this, connId](const std::string& reason) { removeConnection(connId, reason); });
        chPtr->enableReading();
    }

    void StreamServer::removeConnection(types::connection_id_t connId, const std::string& disconnectedReason)
    {
        auto iter = clientChs_.find(connId);
        if (iter != clientChs_.end()) {  // 防止消息重复发送  read error 和 poll 都触发了
            TRACE_INFO << " engine id:"_s << loop_->id() << " connection disconnected, connid:"_s << connId;
            QUEUE_MGR.pubIPCDataToBusiThread(disconnectedReason.c_str(),
                                             static_cast<uint32_t>(disconnectedReason.size()),
                                             shared::InnerMsgId::ON_DISCONNECTED, loop_->id(), connId);
            iter->second->disableAll();
            iter->second->remove();
            clientChs_.erase(connId);
        }
        else {
            TRACE_INFO << "client disconnect but not founnd connId, fd = "_s << connId;
        }
    }

    void StreamServer::createSubServer()
    {
        for (auto* loop : subServerLoops_) {
            auto subserver = std::make_shared<SubStreamServer>(loop);
            subServers_.push_back(subserver);
        }
    }

}  // namespace foundation
}  // namespace xncc